package main

import (
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"strconv"
	"sync"
	"time"
)
//Simple
//连接信息amqp://huangyuzhou:huangyuzhou@127.0.0.1:5672/huangyuzhou这个信息是固定不变的amqp://事固定参数后面两个是用户名密码ip地址端口号Virtual Host
const MQURL = "amqp://hyz:123456@192.168.0.105:5672/hyz"
type RabbitMQ struct {
	conn *amqp.Connection
	channel *amqp.Channel
	//队列名称
	QueueName string
	//交换机名称
	Exchange string
	//key 名称
	Key string
	//连接信息
	Mqurl string
}
//创建实例
func NewRabbitMQ(queueName string, exchange string, key string)*RabbitMQ  {
	return &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, Mqurl: MQURL}
}

//断开channel 和 connection
func (r*RabbitMQ)Destory()  {
	r.channel.Close()
	r.conn.Close()
}
//错误处理函数
func (r *RabbitMQ)failOnErr(err error,message string)  {
	if err !=nil{
		 log.Fatal(err,message)

	}
}
//创建简单模式下RabbitMQ实例
func NewRabbitMQSimple(queueName string)*RabbitMQ  {
	rabbitmq := NewRabbitMQ(queueName,"","")
	var err error
	//获取connection
	rabbitmq.conn,err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err,"failed to connect rabb"+
		"itmq!")
	//获取channel
	rabbitmq.channel,err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}

//直接模式队列生产
func (r*RabbitMQ)PublishSimple(message string)  {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	_,err :=r.channel.QueueDeclare(
			r.QueueName,
		    //是否持久化
			false,
			//是否自动删除
			false,
			//是否具有排他性
			false,
			//是否阻塞处理
			false,
		      //额外的属性
			nil,
		)
	if err !=nil {
		log.Fatal(err,"队列创建失败")
	}
	//调用channel 发送消息到队列中
	r.channel.Publish(
		r.Exchange,
		r.QueueName,
		//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false,
		//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		false,
		amqp.Publishing{
			ContentType:   "text/plain" ,
			Body:            []byte(message),
		})
}
//simple 模式下消费者
func (r*RabbitMQ)ConsumeSimple()  {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	q,err := r.channel.QueueDeclare(
		r.QueueName,
		//是否持久化
		false,
		//是否自动删除
		false,
		//是否具有排他性
		false,
	    //是否阻塞处理
	     false,
	    //额外的属性
		    nil,
		)
	if err !=nil {
		log.Fatal(err,"simple创建队列失败")
	}
	//接收消息
	msg,err := r.channel.Consume(
		q.Name, // queue
		//用来区分多个消费者
		"", // consumer
		//是否自动应答
		true,
		//是否独有
		false,
		//设置为true，表示 不能将同一个Conenction中生产者发送的消息传递给这个Connection中 的消费者
		false, // no-local
		//列是否阻塞
		false, // no-wait
		nil,   // args
		)
	if err !=nil {
		log.Fatal(err)
	}
	forever  := make(chan bool)
	//启动协程处理消息
	go func() {
		for d := range msg{
			//逻辑处理体
			log.Printf("Received a message: %s", d.Body,d.MessageId)
		}

	}()
	log.Printf("消息等待中")
	<- forever
}


/**
Publish模式（订阅模式，消息被路由投递给多个队列，一个消息被多个消费者获取）
 */
//订阅模式创建RabbitMq 实例
func NewRabbitMQPubSub(exchageName string) *RabbitMQ {
    rabbitmq := NewRabbitMQ("",exchageName,"")
    var err error
    //获取conn
    rabbitmq.conn,err = amqp.Dial(rabbitmq.Mqurl)
   	rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
    //获取channel
    rabbitmq.channel,err = rabbitmq.conn.Channel()
    rabbitmq.failOnErr(err,"failed to open a channel")
    return rabbitmq
}
//订阅模式生产
func (r*RabbitMQ)PublishPub(message string)  {
	//尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"fanout",
		true,
		false,
		//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an excha"+
		"nge")
	//2.发送消息
	r.channel.Publish(
		r.Exchange,
		"",
		false,
		false,
		amqp.Publishing{
			ContentType:     "text/plain",
			Body:            []byte(message),
		})
}
//订阅模式消费端代码
func (r*RabbitMQ) RecieveSub() {
	//1.试探性创建交换机
	err :=r.channel.ExchangeDeclare(
		r.Exchange,
		//交换机类型
		"fanout",
		true,
		false,
		//YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		nil,
		)
	r.failOnErr(err,"Failed to declare an exch"+
		"ange")
	//2.试探性创建队列，这里注意队列名称不要写
	q, err := r.channel.QueueDeclare(
		"", //随机生产队列名称
		false,
		false,
		true,
		false,
		nil,
		)
	r.failOnErr(err, "Failed to declare a queue")
	//绑定队列到 exchange 中
	r.channel.QueueBind(
		q.Name,
		//在pub/sub模式下，这里的key要为空
		"",
		r.Exchange,
		false,
		nil,
		)
	//消费消息
	messges, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
		)
	forover := make(chan bool)
	go func() {
		for d := range messges{
			log.Printf("Received a message: %s", d.Body)
		}
	}()
	 <- forover
}


/**
Routing模式(路由模式，一个消息被多个消费者获取，并且消息的目标队列可被生产者指定)
 */
//路由模式
//创建RabbitMQ实例
func NewRabbitMQRouting(exchangeName string, routingKey string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("",exchangeName,routingKey)
	var err error
	rabbitmq.conn,err =amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err,"failed to connect rabbitmq!")
	//获取channel
	rabbitmq.channel,err =rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err,"failed to open a channel")
	return rabbitmq
}
//路由模式发送消息
func (r*RabbitMQ)PublishRouting(message string)  {
	//尝试创建交换机
	err :=  r.channel.ExchangeDeclare(
		r.Exchange,
		//要改成direct
		"direct",
		true,
		false,
		false,
		false,
		nil,
		)
	r.failOnErr(err,"Failed to declare an excha"+
		"nge")
	//发送消息
	r.channel.Publish(
		r.Exchange,
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType:      "text/plain",
			Body:            []byte(message),
		})
}
//路由模式接受消息
func (r *RabbitMQ) RecieveRouting()  {
	//1.试探性创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		//交换机类型
		"direct",
		true,
		false,
		false,
		false,
		nil,
		)
	r.failOnErr(err, "Failed to declare an exch"+
		"ange")
	//2.试探性创建队列，这里注意队列名称不要写
	q, err :=r.channel.QueueDeclare(
		"", //随机生产队列名称
		false,
		false,
		true,
		false,
		nil,
		)
	r.failOnErr(err, "Failed to declare a queue")

	//绑定队列到 exchange 中
	r.channel.QueueBind(
		q.Name,
		//需要绑定key
		r.Key,
		r.Exchange,
		false,
		nil)
	//消费消息
	messges, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	forever := make(chan bool)

	go func() {
		for d := range messges {
			log.Printf("Received a message: %s", d.Body)
		}
	}()

	<-forever
}
/**
topic 模式 （话题模式，一个消息被多个消费者获取，消息的目标queue可用BindingKey以通配符，（#：一个或多个词，*：一个词）的方式指定）
 */
func NewRabbitMQTopic(exchangeName string, routingKey string) *RabbitMQ {
	//创建RabbitMQ实例
	rabbitmq :=NewRabbitMQ("",exchangeName,routingKey)
	var err error
	//获取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
	//获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}
//话题模式发送消息
func (r*RabbitMQ) PublishTopic(message string) {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		//要改成topic
		"topic",
		true,
		false,
		false,
		false,
		nil,
	)

	r.failOnErr(err, "Failed to declare an excha"+
		"nge")

	//2.发送消息
	err = r.channel.Publish(
		r.Exchange,
		//要设置
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

//话题模式接受消息
//要注意key,规则
//其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）
//匹配 kuteng.* 表示匹配 kuteng.hello, kuteng.hello.one需要用kuteng.#才能匹配到
func (r *RabbitMQ) RecieveTopic() {
	//1.试探性创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		//交换机类型
		"topic",
		true,
		false,
		false,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exch"+
		"ange")
	//2.试探性创建队列，这里注意队列名称不要写
	q, err := r.channel.QueueDeclare(
		"", //随机生产队列名称
		false,
		false,
		true,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")

	//绑定队列到 exchange 中
	err = r.channel.QueueBind(
		q.Name,
		//在pub/sub模式下，这里的key要为空
		r.Key,
		r.Exchange,
		false,
		nil)

	//消费消息
	messges, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)

	forever := make(chan bool)

	go func() {
		for d := range messges {
			log.Printf("Received a message: %s", d.Body)
		}
	}()
	<-forever
}

var wg sync.WaitGroup
func main() {
	//simple,work生产者
	//wg.Add(0)
	//for   {
	//	rabbitmq := NewRabbitMQSimple("" +
	//		"hyz")
	//	rabbitmq.PublishSimple("Hello kuteng222!")
	//	wg.Wait()
	//	fmt.Println("发送成功！")
	//
	//}
	//wg.Done()

	//simple,work消费者
	//rabbitmq := NewRabbitMQSimple("" +
	//	"hyz")
	//rabbitmq.ConsumeSimple()
	//fmt.Println("消费成功！")

	//Publish模式生产者
	rabbitmq := NewRabbitMQPubSub(""+ "newProduct")
	for i:=0; i<100;i++  {
		rabbitmq.PublishPub("订阅模式生产第" +strconv.Itoa(i)+ "条数据")
		fmt.Println("订阅模式生产第" +
			strconv.Itoa(i) + "条" + "数据")
		time.Sleep(1 * time.Second)
	}
	//fmt.Println("消费开始！")
	//rabbitmq.RecieveSub()
	//fmt.Println("消费成功！")
	//Routing 路由模式
	//kutengone := NewRabbitMQRouting("kuteng", "kuteng_one")
	//kutengtwo := NewRabbitMQRouting("kuteng", "kuteng_two")
	//for i := 0; i <= 100; i++ {
	//	kutengone.PublishRouting("Hello kuteng one!" + strconv.Itoa(i))
	//	kutengtwo.PublishRouting("Hello kuteng Two!" + strconv.Itoa(i))
	//	time.Sleep(1 * time.Second)
	//	fmt.Println(i)
	//}

	//kutengone :=NewRabbitMQRouting("kuteng", "kuteng_one")
	//kutengone.RecieveRouting()

	//kutengOne := NewRabbitMQTopic("exKutengTopic", "kuteng.topic.one")
	//	//kutengTwo := NewRabbitMQTopic("exKutengTopic", "kuteng.topic.two")
	//	//for i := 0; i <= 100; i++ {
	//	//	kutengOne.PublishTopic("Hello kuteng topic one!" + strconv.Itoa(i))
	//	//	kutengTwo.PublishTopic("Hello kuteng topic Two!" + strconv.Itoa(i))
	//	//	time.Sleep(1 * time.Second)
	//	//	fmt.Println(kutengTwo)
	//	//}

	//
	//kutengOne := NewRabbitMQTopic("exKutengTopic", "#")
	//kutengOne.RecieveTopic()
}